package me.xhy.java.springboot3.webflux.projectreactor.s3projectreactor;

import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

import java.util.concurrent.TimeUnit;

public class S5ThreadAndSchedule {

  // 线程 和 调度
  public static void main(String[] args) throws InterruptedException {

    S5ThreadAndSchedule obj = new S5ThreadAndSchedule();

    /*obj.a();
    obj.b();
    obj.c();
    obj.d();*/
    obj.e();

    TimeUnit.SECONDS.sleep(20);
  }

  public void a() {
    // 1. 流的发布 以及 中间操作，默认使用当前线程
    Flux.range(1, 5)
        .log()
        .map(i -> i * 10)
        .log()
        .subscribe();
  }

  public void b() {
    // 2. Schedulers.immediate() 就是默认的的方式
    System.out.println("=== Schedulers.immediate() ===");
    Flux.range(6, 10)
        .publishOn(Schedulers.immediate())
        .log()
        .map(i -> i * 10)
        .log()
        .subscribe();
  }

  public void c() {
    // 3. Schedulers.single() 一个固定的单线程，同时被 publishOn 和 subscribeOn 使用也是同一个线程
    // publishOn 指定发布者线程；subscribeOn 指定订阅者线程。
    System.out.println("=== Schedulers.single() ===");
    Flux.range(11, 15)
        .publishOn(Schedulers.single())
        .log()
        .map(i -> i * 10)
        .log()
        .subscribeOn(Schedulers.single())
        .log()
        .subscribe();
    // 想指定单线程，可以使用  Schedulers.newSingle(threadName)
  }

  public void d() {
    // 4. Schedulers.boundedElastic() 一个线程数量参照CPU核数确定的 线程池，默认是 cpu核心数*10，队列长度默认 100K
    System.out.println("=== Schedulers.boundedElastic() ===");
    Flux.range(16, 20)
        .publishOn(Schedulers.boundedElastic())
        .log()
        .map(i -> i * 10)
        .log()
        .subscribeOn(Schedulers.single())
        .log()
        .subscribe();
  }

  public void e() {
    // 5. Schedulers.parallel() ForkJoinPoll 线程池
    // 发布者没有指定线程池时，会使用订阅者的线程池
    System.out.println("=== 5. Schedulers.parallel() ===");
    Flux<Integer> flux1 = Flux.range(21, 25)
        .log()
        .map(i -> i * 10)
        .log();

    new Thread(() -> {
      flux1.subscribe();
    }).start();
  }

}
